package fun.tan90.easy.binlog.manager.server;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.lang.Tuple;
import cn.hutool.json.JSONUtil;
import fun.tan90.easy.binlog.base.constants.EasyBinlogConstants;
import fun.tan90.easy.binlog.base.model.OpenBinlogClientReq;
import fun.tan90.easy.binlog.base.model.OpenBinlogClientRes;
import fun.tan90.easy.binlog.base.model.WorkerSystemResource;
import fun.tan90.easy.binlog.base.strategy.AbstractStrategyChoose;
import fun.tan90.easy.binlog.base.threadpool.EasyBinlogThreadPool;
import fun.tan90.easy.binlog.manager.event.DamiBusEvent;
import fun.tan90.easy.binlog.manager.property.EasyBinlogManagerProperties;
import fun.tan90.easy.binlog.manager.ws.WsServer;
import lombok.Getter;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.noear.dami.Dami;
import org.noear.socketd.SocketD;
import org.noear.socketd.transport.core.Entity;
import org.noear.socketd.transport.core.Session;
import org.noear.socketd.transport.core.entity.StringEntity;
import org.noear.socketd.transport.core.listener.EventListener;
import org.noear.socketd.transport.server.Server;
import org.noear.socketd.utils.RunUtils;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * description TODO
 * company 铁人科技
 *
 * @author 陈浩杰
 * @date 2023/12/15 9:48
 */
@Slf4j
@Getter
@Component
public class EbManager implements ApplicationListener<ApplicationReadyEvent>, DisposableBean {

    private final AtomicBoolean initialized = new AtomicBoolean(false);

    public static final String SESSION_ATTR_KEY_WORKER_HAVE_MASTER_CODE = "worker_have_master_code_set";

    public static final String SESSION_ATTR_KEY_WORKER_CONNECTED_TIME = "connected_time";

    private static final Map<String, Session> WORKERS = new HashMap<>();

    private Server tcpServer;

    @Resource
    WsServer wsServer;

    @Resource
    Environment environment;

    @Resource
    AbstractStrategyChoose strategyChoose;

    @Resource
    EasyBinlogManagerProperties easyBinlogManagerProperties;

    @SneakyThrows
    @Override
    public void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) {
        if (initialized.compareAndSet(false, true)) {
            Integer port = environment.getProperty("server.port", Integer.class, 8080);
            String token = easyBinlogManagerProperties.getToken();
            this.tcpServer = SocketD.createServer("sd:tcp")
                    .config(c -> c.port(port + 1).requestTimeout(easyBinlogManagerProperties.getRequestTimeout()))
                    .listen(new EventListener()
                            .doOnOpen(session -> {
                                log.info("EbManager doOnOpen workerId [{}] handshake [{}]", workId(session), JSONUtil.toJsonStr(session.handshake().paramMap()));
                                if (StringUtils.hasLength(token) && !token.equals(session.param("token"))) {
                                    session.close();
                                    return;
                                }
                                registerWorker(session);
                                recordWorkerEventRecord(session, "doOnOpen", "");
                            })
                            .doOnMessage((session, message) -> {
                                String event = message.event();
                                String data = message.dataAsString();
                                boolean request = message.isRequest();
                                boolean subscribe = message.isSubscribe();
                                log.info("EbManager doOnMessage workerId [{}] event [{}] data-size [{}] request [{}] subscribe [{}]", workId(session), event, message.dataSize(), request, message.isSubscribe());
                                if (request || subscribe) {
                                    Entity entity = strategyChoose.chooseAndExecuteResp(event, data);
                                    session.replyEnd(message, entity);
                                } else {
                                    CompletableFuture.runAsync(() -> strategyChoose.chooseAndExecute(event, data),
                                            EasyBinlogThreadPool.EB_FIXED_EXECUTOR);
                                }
                            })
                            .doOnClose(session -> {
                                log.info("EbManager doOnClose {} ", workId(session));
                                unRegisterWorker(session);
                                recordWorkerEventRecord(session, "doOnClose", "");
                            })
                            .doOnError((session, error) -> {
                                log.error("EbManager doOnError {} ", workId(session), error);
                                recordWorkerEventRecord(session, "doOnError", error.getMessage());
                            }))
                    .start();
        }
    }

    @Override
    public void destroy() throws Exception {
        for (Session session : WORKERS.values()) {
            RunUtils.runAndTry(session::closeStarting);
        }
        log.info("EbManager is destroying");
        RunUtils.runAndTry(() -> Thread.sleep(5000));
        for (Session session : WORKERS.values()) {
            RunUtils.runAndTry(session::close);
        }
        tcpServer.stop();
        log.info("EbManager destroyed");
    }

    /**
     * 获取工作节点会话信息
     *
     * @return
     */
    public Map<String, Session> workers() {
        return WORKERS;
    }

    /**
     * 获取worker上的masterCode
     *
     * @return
     */
    public Map<String, Set<String>> workerRunningMasterCode() {
        Map<String, Set<String>> res = new HashMap<>();
        for (Map.Entry<String, Session> next : WORKERS.entrySet()) {
            String workId = next.getKey();
            Session session = next.getValue();
            try {
                String masterCodes = session.sendAndRequest(EasyBinlogConstants.WORKER_RUNNING_MASTER_CODE, null).await().dataAsString();
                res.put(workId, StringUtils.hasLength(masterCodes) ? Stream.of(masterCodes.split(",")).collect(Collectors.toSet()) : new HashSet<>());
            } catch (IOException e) {
                log.error("【{}】发生异常【{}】", workId, e.getMessage());
            }
        }
        return res;
    }

    /**
     * 获取worker系统资源
     *
     * @return
     */
    public WorkerSystemResource workerSystemResource(String workId) {
        Session session = WORKERS.get(workId);
        if (Objects.isNull(session) || !session.isValid()) {
            return WorkerSystemResource.builder().build();
        }
        try {
            String res = session.sendAndRequest(EasyBinlogConstants.WORKER_SYSTEM_RESOURCE, null).await().dataAsString();
            return JSONUtil.toBean(res, WorkerSystemResource.class);
        } catch (IOException e) {
            return WorkerSystemResource.builder().build();
        }
    }

    /**
     * 开启slave binlog
     *
     * @param openBinlogClientReq
     * @return
     */
    public OpenBinlogClientRes openBinlogClient(OpenBinlogClientReq openBinlogClientReq) {
        String masterCode = openBinlogClientReq.getMasterCode();
        Integer minTurnOnBinlogSize = Integer.MAX_VALUE;
        String selectWorkId = null;
        Map<String, Set<String>> workerHaveMasterCodeMap = workerRunningMasterCode();
        for (Map.Entry<String, Set<String>> entry : workerHaveMasterCodeMap.entrySet()) {
            String key = entry.getKey();
            Set<String> masterCodes = entry.getValue();
            if (CollectionUtils.isEmpty(masterCodes)) {
                selectWorkId = key;
                break;
            } else if (!masterCodes.contains(masterCode)) {
                if (minTurnOnBinlogSize.compareTo(masterCodes.size()) >= 0) {
                    minTurnOnBinlogSize = masterCodes.size();
                    selectWorkId = key;
                }
            } else {
                return OpenBinlogClientRes.builder().message("slave客户端已在" + key + "工作节点上开启").build();
            }
        }

        if (!StringUtils.hasLength(selectWorkId)) {
            log.warn("未能找到负载最小的工作节点");
            return OpenBinlogClientRes.builder().message("未能找到负载最小的工作节点").build();
        }

        Session session = WORKERS.get(selectWorkId);
        if (Objects.isNull(session)) {
            return OpenBinlogClientRes.builder().message("与工作节点通讯会话不存在").build();
        }
        if (session.isValid()) {
            try {
                String res = session.sendAndRequest(EasyBinlogConstants.TURN_ON_BINLOG, new StringEntity(JSONUtil.toJsonStr(openBinlogClientReq))).await().dataAsString();
                String message;
                if ("00".equals(res)) {
                    message = "与主库建立连接池失败";
                } else if ("10".equals(res)) {
                    message = "开启binlog同步失败";
                } else {
                    if (!session.attrHas(SESSION_ATTR_KEY_WORKER_HAVE_MASTER_CODE)) {
                        Set<String> binlogClientMasterCode = new HashSet<>();
                        binlogClientMasterCode.add(masterCode);
                        session.attrPut(SESSION_ATTR_KEY_WORKER_HAVE_MASTER_CODE, binlogClientMasterCode);
                    } else {
                        Set<String> binlogClientMasterCode = session.attr(SESSION_ATTR_KEY_WORKER_HAVE_MASTER_CODE);
                        binlogClientMasterCode.add(masterCode);
                    }
                    message = "ok";
                }
                return OpenBinlogClientRes.builder().workId(selectWorkId).message(message).build();
            } catch (IOException e) {
                return OpenBinlogClientRes.builder().workId(selectWorkId).message("与工作节点通讯异常").build();
            }
        }
        return OpenBinlogClientRes.builder().workId(selectWorkId).message("与工作节点通讯会话无效").build();
    }

    /**
     * 关闭slave binlog
     *
     * @param workId
     * @param masterCode
     * @return
     */
    public boolean closeBinlogClient(String workId, String masterCode) {
        Session session = WORKERS.get(workId);
        if (Objects.isNull(session)) {
            return false;
        }
        if (session.isValid()) {
            try {
                String res = session.sendAndRequest(EasyBinlogConstants.TURN_OFF_BINLOG, new StringEntity(masterCode)).await().dataAsString();
                if ("1".equals(res)) {
                    if (session.attrHas(SESSION_ATTR_KEY_WORKER_HAVE_MASTER_CODE)) {
                        Set<String> binlogClientMasterCode = session.attr(SESSION_ATTR_KEY_WORKER_HAVE_MASTER_CODE);
                        binlogClientMasterCode.remove(masterCode);
                    }
                }
                return "1".equals(res);
            } catch (IOException e) {
                return false;
            }
        }
        return false;
    }

    /**
     * 获取主库版本
     *
     * @param workId
     * @param masterCode
     * @return
     */
    public String masterVersion(String workId, String masterCode) {
        Session session = WORKERS.get(workId);
        if (Objects.isNull(session)) {
            return "";
        }
        if (session.isValid()) {
            try {
                return session.sendAndRequest(EasyBinlogConstants.MASTER_VERSION, new StringEntity(masterCode)).await().dataAsString();
            } catch (IOException e) {
                return "";
            }
        }
        return "";
    }

    private String workId(Session session) {
        return session.name();
    }

    private void registerWorker(Session session) throws IOException {
        session.attrPut(SESSION_ATTR_KEY_WORKER_CONNECTED_TIME, DateUtil.now());
        String workId = workId(session);
        if (WORKERS.containsKey(workId)) {
            session.close();
            WORKERS.remove(workId);
        } else {
            WORKERS.put(workId, session);
        }
        EasyBinlogThreadPool.EB_FIXED_EXECUTOR.execute(() -> {
            wsServer.broadcast("online", workId);
            Dami.<Session, Void>bus().send(DamiBusEvent.WORKER_ONLINE, session);
        });
    }

    private void unRegisterWorker(Session session) {
        String workId = workId(session);
        WORKERS.remove(workId);
        EasyBinlogThreadPool.EB_FIXED_EXECUTOR.execute(() -> {
            wsServer.broadcast("offline", workId);
            Dami.<String, Void>bus().send(DamiBusEvent.WORKER_OFFLINE, workId);
        });
    }

    private void recordWorkerEventRecord(Session session, String eventType, String eventContent) {
        String workId = workId(session);
        EasyBinlogThreadPool.EB_FIXED_EXECUTOR.execute(() -> {
            Dami.<Tuple, Void>bus().send(DamiBusEvent.RECORD_WORKER_EVENT_RECORD, new Tuple(workId, eventType, eventContent));
        });
    }
}
